package com.whale.springboot.kafka.java.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

/**
 * kafka的消费者，指定offset 进行消费（升级版，不要用consumer.subscribe，而采用assign）
 */
public class CustomConsumerSeekPro {

    public static void main(String[] args) {

        // 0 配置信息
        Properties properties = new Properties();

        // 连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.179.131:9092");

        // 反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"JKH1");

        // 1 创建消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        // 既然要自己指定一个确定的其实消费位置，那一般的隐含之意就是：
        //      不需要去参与消费组，再重平衡机制，因为我们只要调用consumer.subscribe就会触发再平衡
        // 但是用consumer.subscribe也是可以的，只是不符合别人的设计思想

        // 方式一：只指定一个分区0（可行）
        TopicPartition topicPartition = new TopicPartition("test-more", 0);
        kafkaConsumer.assign(Arrays.asList(topicPartition));
        kafkaConsumer.seek(topicPartition,1); // 指定从当前主题的当前分区的 offset为1的位置开始消费

        // 方式二：找到了之后直接for循环，然后seek（不可行）
        // 草，这个做法是错的，因为seek之后，kafkaConsumer只会消费最后一个seek指定的分区去消费，导致我下面poll的时候只能拿一个分区，shit!!!
//        List<PartitionInfo> partitions = kafkaConsumer.partitionsFor("test-more");
//        for (PartitionInfo partition : partitions) {
//            TopicPartition topicPartition = new TopicPartition("test-more", partition.partition());
//            kafkaConsumer.assign(Arrays.asList(topicPartition));
//            kafkaConsumer.seek(topicPartition,1); // 指定从当前主题的当前分区的 offset为1的位置开始消费
//        }

        // 方式三：正确姿势，只能循环，一个分区一个分区的来操作了，只适合用来有结束条件的业务场景，比如，我需要查询指定时间段的消息，那么下面while(true)结束条件就是 record.timestamp>endTime
//        List<PartitionInfo> partitions = kafkaConsumer.partitionsFor("test-more");
//        for (PartitionInfo partition : partitions) {
//            TopicPartition topicPartition = new TopicPartition("test-more", partition.partition());
//            kafkaConsumer.assign(Arrays.asList(topicPartition));
//            kafkaConsumer.seek(topicPartition,1); // 指定从当前主题的当前分区的 offset为1的位置开始消费
//
//            while (true){
//                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
//                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
//                }
//            }
//        }

        // 3  消费数据
        while (true){

            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                // offset
                // topic
                // partition
                // key
                // value
                // timestamp：时间戳
                // timestampType：时间戳类型，   本条记录的创建时间（生产者）、本条记录的追加时间（broker写入log文件的时间）  2选1

                // headers：数据头，生产者在写入数据的时候，可以附加加进来的一些额外数据，这是用户自定义的

                System.out.println(consumerRecord);
            }
        }
    }
}
